package com.yj.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import com.yj.core.exception.ServiceException;
import com.yj.core.result.JsonResult;
import com.yj.rocketmq.MqConstants;
import com.yj.rocketmq.RocketMqConstants;
import com.yj.rocketmq.RocketMqExceptionMsg;
import com.yj.rocketmq.api.entity.MessageRecord;
import com.yj.rocketmq.api.mapper.MessageRecordMapper;
import com.yj.rocketmq.listener.LocalTransactionListener;
import com.yj.rocketmq.producer.dto.MessageDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.util.Iterator;
import java.util.Map;
import java.util.UUID;

/**
 * 发送消息基类
 *
 * @author cyj
 * @date 19-6-24
 */
@Slf4j
public abstract class AbstractMqSendMsgHandler {

    @Autowired
    private LocalTransactionListener transactionListener;

    @Autowired
    private MessageRecordMapper messageRecordMapper;

    /**
     * 顺序消息
     *
     * @param msgDTO:消息体
     * @return
     */
    public void sendP2P(MessageDTO msgDTO) {
        //校验入参
        checkValid(msgDTO);

        msgDTO.setTag(convertTag(msgDTO.getTopic(), msgDTO.getTag()));
        this.sendMsg(msgDTO);
    }

    /**
     * 顺序消息
     *
     * @param msgDTO：消息体
     * @return
     */
    public JsonResult sendBroadcast(MessageDTO msgDTO) {

        checkValid(msgDTO);

        return sendMsg(msgDTO);
    }


    /**
     * 发送事务消息
     *
     * @param msgDTO:消息体
     * @return
     */
    public void sendTransMsg(MessageDTO msgDTO) {
        //校验入参
        checkValid(msgDTO);
        //设置tag
        msgDTO.setTag(convertTag(msgDTO.getTopic(), msgDTO.getTag()));

        MessageRecord messageRecord = new MessageRecord(msgDTO);

        //写入事务中间表
        messageRecordMapper.insert(messageRecord);

        //构造消息
        Message message = buildMessage(msgDTO, messageRecord.getMsgId());

        //发送事务消息
        getMqTransProducer().getProducer().setTransactionListener(transactionListener);
        TransactionSendResult result;
        try {
            result = getMqTransProducer().getProducer().sendMessageInTransaction(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("事务消息发送成功");
                }

                @Override
                public void onException(Throwable e) {
                    log.info("事务消息发送失败");
                    throw new ServiceException(RocketMqExceptionMsg.SEND_MESSAGE_EXCEPTION);
                }
            });
            if (result != null) {
                log.info("{}事务消息发送成功，messageId={}，返回结果={}", RocketMqConstants.LOG_PREFIX,
                        result.getMsgId(), JSON.toJSONString(result));
            }
        } catch (MQClientException e) {
            log.error("{} 事务消息失败消息发送失败，需要进行重试处理", RocketMqConstants.LOG_PREFIX);
            e.printStackTrace();
            throw new ServiceException(RocketMqExceptionMsg.SEND_MESSAGE_FAIL);
        }
    }

    /**
     * 发送mq消息
     *
     * @param msgDTO
     * @return
     */
    private JsonResult sendMsg(MessageDTO msgDTO) {
        String msgId = UUID.randomUUID().toString();
        //构造消息
        Message message = buildMessage(msgDTO,msgId);

        try {
            SendResult result = getMqProducer().getProducer().send(message, (mqs, msg1, pk) -> {
                int select;
                if (pk instanceof Long) {
                    select = ((Long) pk).intValue();
                } else if (pk instanceof Integer) {
                    select = ((Integer) pk).intValue();
                } else {
                    select = Math.abs(pk.toString().hashCode());
                }
                if (select < 0) {
                    select = 0;
                }
                return mqs.get(select % mqs.size());
            }, msgId);

            if (result != null) {
                log.info("{} 有序消息发送成功，messageId={}，返回结果={}", RocketMqConstants.LOG_PREFIX,
                        result.getMsgId(), JSON.toJSONString(result));
                return new JsonResult(result);
            }

            return new JsonResult(RocketMqExceptionMsg.SEND_MESSAGE_FAIL);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("{} apacheRocketMq 有序消息发送失败，需要进行重试处理", RocketMqConstants.LOG_PREFIX);
            return new JsonResult(RocketMqExceptionMsg.SEND_MESSAGE_EXCEPTION);
        }
    }


    /**
     * 构造mq消息
     *
     * @param msgDTO
     * @return
     */
    private Message buildMessage(MessageDTO msgDTO, String msgId) {
        /**
         * 发送内容
         */
        String msgContent = msgDTO.getBody();
        if (StringUtils.isBlank(msgContent)) {
            msgContent = "";
        }

        /**
         * 构造发送消息
         */
        Message message = new Message(msgDTO.getTopic().trim().toUpperCase(),
                msgDTO.getTag().trim().toUpperCase(),
                msgId,
                msgContent.getBytes());

        //消息扩展属性
        if (!CollectionUtils.isEmpty(msgDTO.getProperties())) {
            setUserProperty(msgDTO.getProperties(), message);
        }
        //延迟参数
        if (msgDTO.getDelayLevel() != null
                && !msgDTO.getDelayLevel().equals(0)) {
            message.setDelayTimeLevel(msgDTO.getDelayLevel());
        }
        return message;
    }

    /**
     * 转换tag
     *
     * @return
     */
    private String convertTag(String topic, String srcTag) {
        MqConstants.MQ_TOPIC mqTopic = MqConstants.MQ_TOPIC.getMqTopic(topic);

        if (mqTopic.getBroadCast()) {
            return srcTag;
        }
        return srcTag;
    }


    /**
     * 设置用参数
     *
     * @param properties
     * @param message
     */
    private void setUserProperty(Map<String, String> properties, Message message) {
        Iterator<Map.Entry<String, String>> iterator = properties.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, String> entry = iterator.next();
            if (StringUtils.isNotBlank(entry.getKey())) {
                message.putUserProperty(entry.getKey(), entry.getValue());
            }
        }
    }


    /**
     * 校验参数
     *
     * @param messageDTO
     * @return
     */
    private void checkValid(MessageDTO messageDTO) {
        if (messageDTO == null) {
            log.error(RocketMqExceptionMsg.PARAM_IS_NULL.getMessage());
            throw new ServiceException(RocketMqExceptionMsg.PARAM_IS_NULL);
        }

        if (StringUtils.isBlank(messageDTO.getTopic())) {
            log.error(RocketMqExceptionMsg.TOPIC_NOT_NULL.getMessage());
            throw new ServiceException(RocketMqExceptionMsg.TOPIC_NOT_NULL);
        }
        if (StringUtils.isBlank(messageDTO.getTag())) {
            log.error(RocketMqExceptionMsg.TAG_NOT_NULL.getMessage());
            throw new ServiceException(RocketMqExceptionMsg.TAG_NOT_NULL);
        }
        if (StringUtils.isBlank(messageDTO.getBody())) {
            log.error(RocketMqExceptionMsg.BODY_NOT_NULL.getMessage());
            throw new ServiceException(RocketMqExceptionMsg.BODY_NOT_NULL);
        }
    }

    /**
     * 校验参数
     *
     * @param topic
     * @param msgId
     * @param body
     * @param tag
     * @return
     */
    private JsonResult checkValid(String topic, String msgId, String body, String tag) {

        return new JsonResult();
    }


    /**
     * 获取发送消息句柄
     *
     * @return
     */
    protected abstract AbstractMqProducer getMqProducer();

    /**
     * 获取发送消息句柄
     *
     * @return
     */
    protected abstract AbstractMqTransProducer getMqTransProducer();


}
